package pku;

import java.io.*;
import java.util.HashMap;
import java.util.List;

public class MessageStore {
    static final MessageStore store = new MessageStore();
    File file = new File("./data/data.txt");
    static OutputStream out;
    static InputStream in;
    //给每个consumer对应一个流
    HashMap<String, InputStream> inMap = new HashMap<String, InputStream>();

    // 同步执行push方法，防止多个线程同时访问互斥资源
    public synchronized void push(ByteMessage msg) {
        if (msg == null) {
            return;
        }
        //获取topic
        String topic = msg.headers().getString(MessageHeader.TOPIC);

        try {
            if (out == null)
                out = new FileOutputStream(file, true);
            //写一个byte，表示body和topic的长度+2
            out.write((byte) msg.getBody().length + topic.getBytes().length + 2);
            out.write((byte) topic.getBytes().length);
            out.write(topic.getBytes());

            out.write((byte) msg.getBody().length);
            out.write(msg.getBody());
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public synchronized ByteMessage pull(String queue, List<String> topics) {
        try {
            if (!inMap.containsKey(queue))
                inMap.put(queue, new FileInputStream(file));
            //每个queue都有一个InputStream
            in = inMap.get(queue);
            if (in.available() == 0) {
                return null;
            }
            byte[] byteTopic;
            byte[] body;
            //每次循环读一个message的数据量
            String ss = "";
            do {
                byte lenTotal = (byte) in.read();
                //读到文件尾了，则lenTotal为-1
                if (lenTotal == -1)
                    return null;
                byte[] byteTotal = new byte[lenTotal];
                in.read(byteTotal);

                byte lenTopic = byteTotal[0];
                byteTopic = new byte[lenTopic];
                System.arraycopy(byteTotal, 1, byteTopic, 0, lenTopic);

                byte lenBody = byteTotal[lenTopic + 1];
                body = new byte[lenBody];
                //读取body部分内容，存入body数组
                System.arraycopy(byteTotal, lenTopic + 2, body, 0, lenBody);

                ss = new String(byteTopic);
            } while (!topics.contains(ss));
            ByteMessage msg = new DefaultMessage(body);
            msg.putHeaders(MessageHeader.TOPIC, ss);
            return msg;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;

    }
}
